#pragma once
#include <iostream>
#include "../mqcommon/Helper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <mutex>
#include <memory>
#include <list>
#include <unordered_map>
namespace mq_zxws
{
    using MessagePtr = std::shared_ptr<mq_zxws::Message>;
    #define DATAFILE_SUFFIX ".mqd"
    #define TMPFILE_SUFFIX ".mqd.tmp"
    class MessageMapper{
    public:
        MessageMapper(std::string& basedir, const std::string& qname)
            :_qname(qname)
        {
            if(basedir.back() != '/') basedir += '/';
            _datafile = basedir + qname + DATAFILE_SUFFIX;
            _tmpfile = basedir + qname + TMPFILE_SUFFIX;
            assert(FileHelper::createDirectory(basedir));
            createMsgFile();
        }
        bool createMsgFile(){
            bool ret = FileHelper::createFile(_datafile);
            if(ret == false){
                ELOG("创建 %s 文件失败",_datafile.c_str());
                return false;
            }
            return true;
        }
        void removeMsgFile(){
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool insert(MessagePtr & msg){
            return insert(_datafile, msg);
        }
        bool remove(const MessagePtr& msg){
            //1、将有效位置零
            msg->mutable_payload()->set_valid("0");
            //1、序列化，获取数据内容
            std::string body = msg->payload().SerializeAsString();
            if(body.size() != msg->length()){
                DLOG("队列中的数据不一致");
                return false;
            }
            //2、获取文件长度
            FileHelper helper(_datafile);
            //3、将数据写入指定位置
            if(!helper.write(body.c_str(), msg->offset(), body.size())){
                DLOG("向队列文件中写入数据失败");
                return false;
            }
            return true;
        }
        std::list<MessagePtr> gc(){
            std::list<MessagePtr> result;
            bool ret = true;
            //1、加载有效数据
            ret = load(result);
            if(!ret){
                DLOG("加载数据失败");
                return result;
            }
            //2、将有效数据写入临时文件
            FileHelper::createFile(_tmpfile);
            for(auto & msg: result){
                ret = insert(_tmpfile, msg);
                if(!ret){
                    DLOG("向临时文件写入数据出错");
                    return result;
                }
            }
            
            //3、删除旧文件，将临时文件改名
            ret = FileHelper::removeFile(_datafile);
            if(!ret){
                DLOG("删除旧文件出错");
                return result;
            }

            ret = FileHelper(_tmpfile).rename(_datafile);
            if(!ret){
                DLOG("临时文件改名出错");
                return result;
            }
            for(auto ptr : result){
                std::cout << ptr->payload().body() << std::endl;
            }
            return result;
        }
    private:
        bool load(std::list<MessagePtr>& result){
            bool ret = true;
            FileHelper helper(_datafile);
            size_t fsize = helper.size();
            size_t offset = 0;
            while(offset < fsize){
                size_t len = 0;
                ret = helper.read((char*)&len, offset, sizeof(size_t));
                if(!ret){
                    DLOG("读取数据大小失败");
                    return false;
                }
                offset += sizeof(size_t);
                std::string body(len, 0);
                ret = helper.read(&body[0], offset, len);
                if(!ret){
                    DLOG("读取数据失败");
                    return false;
                }
                offset += len;
                MessagePtr ptr = std::make_shared<Message>();
                ptr->mutable_payload()->ParseFromString(body);
                if(ptr->payload().valid() != "0"){
                    result.push_back(ptr);
                }
            }
            return true;
        }
        bool insert(const std::string & filename, MessagePtr & msg){ // 没有存放文件大小?
            //1、序列化，获取数据内容
            std::string body = msg->payload().SerializeAsString();
            //2、获取文件长度
            FileHelper helper(filename);
            size_t sz = helper.size();
            size_t msg_sz = body.size();
            //3、将数据写入指定位置
            if(!helper.write((char*)&msg_sz, sz, sizeof(msg_sz))){
                DLOG("向队列文件中写入数据长度失败");
                return false;
            }
            if(!helper.write(body.c_str(), sz + sizeof(msg_sz), body.size())){
                DLOG("向队列文件中写入数据失败");
                return false;
            }
            //3、修改文件的位置
            msg->set_offset(sz + sizeof(msg_sz));
            msg->set_length(body.size());
            return true;
        }
    private:
        std::string _datafile;
        std::string _tmpfile;
        std::string _qname;
    };

    class QueueMessage{
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string& basedir, const std::string& qname)
            :_qname(qname)
            ,_valid_count(0)
            ,_total_count(0)
            ,_mapper(basedir, qname)
        {}
        void recovery(){
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();
            for(auto & msg : _msgs){
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _valid_count = _total_count = _durable_msgs.size();
        }
        bool insert(const BasicProperties * bp, std::string body, bool queue_is_durable){
            // 创建消息对象
            auto msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if(bp != nullptr){
                DeliverMode mode = queue_is_durable ? bp->deliver_mode() : DeliverMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }else{
                DeliverMode mode = queue_is_durable ? DeliverMode::DURABLE : DeliverMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUID::uuid());
                msg->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            // 是否进行持久化处理
            std::unique_lock<std::mutex> lock(_mutex);
            if(msg->payload().properties().deliver_mode() == DeliverMode::DURABLE){
                msg->mutable_payload()->set_valid("1");
                // 持久化
                if(!_mapper.insert(msg)){
                    DLOG("持久化数据失败 %s", body.c_str());
                    return false;
                }
                _valid_count++;
                _total_count++;
                // 加入持久化hashmap
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 内存管理
            _msgs.push_back(msg);
            return true;
        }
        bool remove(const std::string& msg_id){ // 在删除的同时产看是否需要进行垃圾回收
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _waitack_msgs.find(msg_id);
            if(it == _waitack_msgs.end()) return true;
            auto msg = it->second;
            if(msg->payload().properties().deliver_mode() == DeliverMode::DURABLE){
                if(!_mapper.remove(msg)){
                    DLOG("删除数据失败");
                    return false;
                }
                _valid_count--;
                _durable_msgs.erase(msg_id);
                gc();
            }
            _waitack_msgs.erase(it);
            return true;
        }
        MessagePtr front(){
            std::unique_lock<std::mutex> lock(_mutex);
            if(_msgs.empty()) return MessagePtr();
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // DLOG("取出的数据为%s",msg->payload().body().c_str());
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }
        size_t getable_count(){
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t push_count(){
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        size_t total_count(){
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t durable_count(){
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        size_t waitack_count(){
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void clear(){
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _total_count = _valid_count = 0;
        }
    private:
        bool GCCheck(){
            return _total_count > 2000 && _valid_count * 10 / _total_count < 5;
        }
        void gc(){
            if(GCCheck()){
                // 垃圾回收机制
                auto msgs = _mapper.gc();
                for(auto & msg:msgs){
                    auto it = _durable_msgs.find(msg->payload().properties().id());
                    if(it == _durable_msgs.end()){
                        DLOG("有未被加载到内存中的消息，消息id : %s", msg->payload().properties().id().c_str());
                        _msgs.push_back(msg);
                        _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                        continue;
                    }
                    // 修改信息的位置
                    it->second->set_offset(msg->offset());
                    it->second->set_length(msg->length());
                }
                // 修改有效数据的大小
                _total_count = _valid_count = _durable_msgs.size();
            }
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count; // 持久化数据的有效个数
        size_t _total_count; // 持久化数据的总量
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs; // 待推送消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 需要持久化的消息hashmap [id, ptr]
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 等待确认的消息队列 [id, ptr]
    };

    class MessageManager{
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string& basedir):_basedir(basedir){}
        // 创建队列
        void initQueueMessage(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it != _queue_msgs.end()) return;
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->recovery();
        }
        // 销毁队列
        void destoryQueueMessage(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()) return;
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear();
        }
        // 插入消息
        bool insert(const std::string& qname, const BasicProperties * bp, std::string body, bool delivermode){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("插入消息失败，不存在消息队列 %s", qname.c_str());
                    return true;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body, delivermode);
        }
        // 删除消息
        bool remove(const std::string& qname, const std::string & id){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("删除消息，不存在消息队列 %s", qname.c_str());
                    return true;
                }
                qmp = it->second;
            }
            return qmp->remove(id);
        }
        // 获取队首消息
        MessagePtr front(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                
                if(it == _queue_msgs.end()){
                    DLOG("获取队首元素失败，不存在消息队列 %s", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        // 确认消息
        void ack(const std::string& qname, const std::string & id){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("确认消息失败，不存在消息队列 %s", qname.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(id);
        }

        size_t getable_count(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("getable_count fail，不存在消息队列 %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        size_t push_count(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("push_count fail，不存在消息队列 %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->push_count();
        }
        size_t total_count(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("total_count fail，不存在消息队列 %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        size_t durable_count(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("durable_count fail，不存在消息队列 %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        size_t waitack_count(const std::string& qname){
            // QueueMessage::ptr qmp;
            // {
            //     std::unique_lock<std::mutex> lock(_mutex);
            //     auto it = _queue_msgs.find(qname);
            //     if(it == _queue_msgs.end()){
            //         DLOG("获取队首元素失败，不存在消息队列 %s", qname.c_str());
            //         return;
            //     }
            //     qmp = it->second;
            // }
            // return qmp->waitack_count();
            return getQueuetMessagePtr(qname)->waitack_count();
        }
        void clear(){
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto& msg:_queue_msgs){
                msg.second->clear();
            }
        }
    private:
        QueueMessage::ptr getQueuetMessagePtr(const std::string& qname){
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end()){
                    DLOG("getQueuetMessagePtr fail，不存在消息队列 %s", qname.c_str());
                    return QueueMessage::ptr();
                }
                qmp = it->second;
            }
            return qmp;
        }
        
    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}